package com.bw.gmall.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.bw.gmall.realtime.bean.ShopBean;
import com.bw.gmall.realtime.utils.DateFormatUtil;
import com.bw.gmall.realtime.utils.MyClickHouseUtil;
import com.bw.gmall.realtime.utils.MyKafkaUtil;
import com.bw.gmall.realtime.utils.MyKafkaUtil2;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.sql.Timestamp;
import java.time.Duration;

/*
1.获取ods 中日志数topic_log
2.处理过滤topic_log
3.对新老用户进行处理
               is_news:1
                         状态=null  我要把当前人ts转换成年月日  跟新到状态值
                         状态！=null   判断当前数据的ts 和状态日期是否相同  如果不同跟新数据
               is_new:0
                         状态=null    将ts-1day 跟新到状态中
                         如果有值 不需要处理
4. 把主流数据拆分成5个流   分别存入到5个主题中
1.创建流式环境
2.设置并行度
3.运行jar 向topic_log主题 发送数据
4.从 Kafka 读取主流数据
5.数据清洗，转换结构
6.将脏数据写出到 Kafka 指定主题
 */


public class BaseLogAppShopDws {
    public static void main(String[] args) throws Exception {

        // TODO 1. 初始化环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        String topic = "ods_traffic";
        String groupId = "base_log_consumer";
        String topicuv = "dwd_Shop_uv";
        String groupIduv = "base_log_uv";

        DataStreamSource<String> source = env.addSource(MyKafkaUtil.getFlinkKafkaConsumer(topic, groupId));
        DataStreamSource<String> sourceuv = env.addSource(MyKafkaUtil.getFlinkKafkaConsumer(topicuv, groupIduv));

        //TODO 3.将数据转换为ShopBean对象

        SingleOutputStreamOperator<ShopBean> uvShop = sourceuv.flatMap(new FlatMapFunction<String, ShopBean>() {
            @Override
            public void flatMap(String value, Collector<ShopBean> out) throws Exception {
                JSONObject jsonObject = JSONObject.parseObject(value);
                JSONObject common = jsonObject.getJSONObject("common");

                Long ts = jsonObject.getLong("ts");
                out.collect(new ShopBean(null, null,
                        common.getInteger("shop_id"),
                        common.getInteger("device_type"),
                        1L, 0L, ts
                ));
            }
        });

//        uvShop.print();

        SingleOutputStreamOperator<ShopBean> pvShop = source.flatMap(new FlatMapFunction<String, ShopBean>() {
            @Override
            public void flatMap(String value, Collector<ShopBean> out) throws Exception {
                JSONObject jsonObject = JSONObject.parseObject(value);
                JSONObject common = jsonObject.getJSONObject("common");

                Long ts = jsonObject.getLong("ts");
                out.collect(new ShopBean(null, null,
                        common.getInteger("shop_id"),
                        common.getInteger("device_type"),
                        0L, 1L, ts
                ));
            }
        });

//        pvShop.print();


        //TODO 4.提取事件时间生成Watermark
        SingleOutputStreamOperator<ShopBean> ShopWatermarks = uvShop.union(pvShop).assignTimestampsAndWatermarks(WatermarkStrategy
                .<ShopBean>forBoundedOutOfOrderness(Duration.ofSeconds(5))
                .withTimestampAssigner(new SerializableTimestampAssigner<ShopBean>() {
                    @Override
                    public long extractTimestamp(ShopBean element, long recordTimestamp) {
                        return element.getTs();
                    }
                })
        );

//        ShopWatermarks.map(a->a.getTs()).print("---->");
        //TODO 5.分组

        KeyedStream<ShopBean, Tuple2<Integer, Integer>> ShopKeyBy = ShopWatermarks.keyBy(new KeySelector<ShopBean, Tuple2<Integer, Integer>>() {
            @Override
            public Tuple2<Integer, Integer> getKey(ShopBean value) throws Exception {
                //店铺   PC端
                return Tuple2.of(value.getShop_id(), value.getDevice_type());
            }
        });

//        ShopKeyBy.print();

        //TODO 6.开窗聚合
        SingleOutputStreamOperator<ShopBean> reduceDs = ShopKeyBy.window(TumblingEventTimeWindows.of(Time.seconds(10)))
                .reduce(new ReduceFunction<ShopBean>() {
                    @Override
                    public ShopBean reduce(ShopBean value1, ShopBean value2) throws Exception {
                        value1.setUniqueUserCount(value1.getUniqueUserCount() + value2.getUniqueUserCount());
                        value1.setPv(value1.getPv() + value2.getPv());
                        return value1;
                    }
                }, new ProcessWindowFunction<ShopBean, ShopBean, Tuple2<Integer, Integer>, TimeWindow>() {
                    @Override
                    public void process(Tuple2<Integer, Integer> integerIntegerTuple2, Context context, Iterable<ShopBean> elements, Collector<ShopBean> out) throws Exception {
                        ShopBean shopBean = elements.iterator().next();
                        shopBean.setStt(Timestamp.valueOf(DateFormatUtil.toYmdHms(context.window().getStart())));
                        shopBean.setEdt(Timestamp.valueOf(DateFormatUtil.toYmdHms(context.window().getEnd())));
                        shopBean.setTs(System.currentTimeMillis());
                        out.collect(shopBean);
                    }
                });



        reduceDs.print("---------->");
        reduceDs.addSink(MyClickHouseUtil.getSinkFunction(
                "insert into dws_shop  values(?,?,?,?,?,?,?)"));

        env.execute();
    }
}
